package com.foxchen.offline.task;


import cn.hutool.db.Db;
import com.foxchen.offline.task.bean.TaskPojo;
import com.foxchen.offline.task.config.OfflineTaskConfig;
import com.foxchen.offline.task.enums.OfflineTaskStatusEnum;
import com.foxchen.offline.task.service.OfflineTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author 6774000316@zte.intra
 */
@Slf4j
@Component
public class TaskHandlerFactory {
    @Resource
    private ThreadPoolExecutor exportThreadPoolExecutor;
    @Resource
    private Map<String,ExportHandler> handlerMap = new ConcurrentHashMap<>();
    @Resource
    private OfflineTaskConfig exportConfig;
    @Resource
    private DataSource dataSource;
    @Lazy
    @Resource
    private OfflineTaskService offlineTaskService;

    /**
     * 记录执行中的线程
     */
    private static final Map<String, Future<Boolean>> RUNNING_THREAD = new ConcurrentHashMap<>();

    @PostConstruct
    public void dataSourceParamCheck() throws Exception {
        // 检查是否已经存在数据库，没有则创建
        DefaultResourceLoader defaultResourceLoader = new DefaultResourceLoader(getClass().getClassLoader());
        org.springframework.core.io.Resource resource = defaultResourceLoader.getResource("sql/offline_task.sql");

        try(BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(resource.getInputStream()))){
            StringBuilder sql = new StringBuilder();
            bufferedReader.lines().forEach(lineSql -> {
                sql.append(lineSql);
                sql.append(System.lineSeparator());
            });

            log.info("offline task create sql=\n{}",sql);
            Db.use(dataSource).execute(sql.toString());

            log.info("offline task is create ok.");
        }
    }

    public void executeTask(TaskPojo taskPojo) {
        // 通过任务类型找到导出handler
        ExportHandler exportHandler = handlerMap.get(taskPojo.getType());

        // 没有找到处理类。调用rpc更新任务状态
        if(exportHandler == null) {
            log.warn("task type:{} is not support.",taskPojo.getType());
            taskPojo.setStatus(OfflineTaskStatusEnum.FAILED.getStatus());
            offlineTaskService.updateTask(taskPojo);
            return;
        }


        // 有处理类。调用处理类方法处理
        Future<Boolean> future = exportThreadPoolExecutor.submit(() ->
                exportHandler.dealExport(exportConfig,taskPojo,RUNNING_THREAD,offlineTaskService));
        RUNNING_THREAD.put(taskPojo.getId(), future);
    }

    public void stopTaskByTaskId(String taskId){
        // 从运行中的map 拿到执行中的线程
        Future<Boolean> future = RUNNING_THREAD.get(taskId);

        // 如果存在 直接停止运行
        if(future != null){
            try {
                future.cancel(Boolean.TRUE);
            }
            catch (Exception e) {
                log.error("task stop error.task id:{}",taskId,e);
            }
            finally {
                RUNNING_THREAD.remove(taskId);
            }
        }
    }
}
